package com.example.reactor;

import org.reactivestreams.Subscription;
import reactor.core.Disposable;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

import java.io.IOException;
import java.time.Duration;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class FluxDemo5 {

    public static void main(String[] args) throws IOException {
//        buffer();
//        limit();
//        generate();
//        dispose();
//        create();
//        handle();
//        thread();
        threadTest();
        System.in.read();
    }

    /**
     * 发布者默认使用订阅者线程池
     */
    public static void threadTest() {
        Scheduler threadTest = Schedulers.newParallel("threadTest", 4);
        Flux<String> flux = Flux.range(1, 5)
                .log()
                .map(integer -> "数据：" + integer)
                .publishOn(threadTest)
                .map(string -> "数据->" + string)
                .log();
        new Thread(
                () -> {
//                    flux.subscribe(value -> System.out.println(value + "->" + Thread.currentThread().getName()));
                    flux.subscribe(new BaseSubscriber<String>() {
                        @Override
                        protected void hookOnSubscribe(Subscription subscription) {
                            System.out.println(Thread.currentThread().getName() + "钩子函数，订阅关系" + subscription);
                            request(1);
                        }

                        @Override
                        protected void hookOnNext(String value) {
                            System.out.println(Thread.currentThread().getName() + "钩子函数，接收数据：" + value);
                            request(1);
                        }

                        @Override
                        protected void hookOnComplete() {
                            System.out.println(Thread.currentThread().getName() + "钩子函数，数据流完成");
                        }
                    });
                }
        ).start();
    }

    /**
     * 切换线程池
     */
    public static void thread() {
        Flux.range(1, 10)
                /**
                 * 改变发布者的线程池
                 */
                .publishOn(Schedulers.fromExecutor(
                        new ThreadPoolExecutor(4, 8, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(1024))
                ))
                .log()
                .map(integer -> integer * 10)
                .log()
                /**
                 * 改变订阅者的线程池
                 */
                .subscribeOn(Schedulers.single())
                .subscribe();
        /**
         * 调度器
         */
//        Schedulers.immediate(); // 默认调度器
//        Schedulers.single(); // 单线程调度器
//        Schedulers.boundedElastic(); // 弹性调度器
//        Schedulers.fromExecutor(
//                new ThreadPoolExecutor(4, 8, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(1024))
//        ); // 自定义调度器
    }

    /**
     * handle 是 Project Reactor 中 Flux 和 Mono 的一个操作符，其主要作用如下：
     * 1、数据转换与过滤：
     * handle 允许你对流中的每个元素进行处理，并可以决定是否以及如何发出数据
     * 你可以使用 SynchronousSink 来发出零个、一个或多个元素
     * 2、错误处理：
     * 在 handle 中可以直接通过 synchronousSink.error() 发出错误信号
     * 也可以通过不调用 next() 来跳过某些元素（相当于过滤）
     * 3、与 map 和 filter 的区别：
     * 相比于 map（只能一对一转换）和 filter（只能过滤），handle 更加灵活
     * 它可以在一个操作中同时完成转换和过滤的功能
     * 4、同步处理：
     * handle 是同步操作，适用于需要逐个处理元素的场景
     */
    public static void handle() {
        Flux.range(1, 10)
                /**
                 * SynchronousSink 是 Project Reactor 中的一个重要接口，其主要作用如下：
                 * 1、数据发射控制：
                 * 允许在操作符中精确控制数据的发射
                 * 可以选择发射一个、多个或零个元素
                 * 提供 next() 方法来发射数据
                 * 2、流的终止控制：
                 * 提供 complete() 方法来正常终止流
                 * 提供 error() 方法来发出错误信号并终止流
                 * 3、使用场景限制：
                 * 只能在当前操作符的执行上下文中同步使用
                 * 不能在异步回调中使用（需要使用 FluxSink）
                 * 4、主要使用位置：
                 * 在 handle() 操作符中作为参数
                 * 在 Flux.generate() 方法中作为参
                 */
                .handle(
                        (integer, synchronousSink) -> {
                            synchronousSink.next("数据1：" + integer);
//                            synchronousSink.next("数据2：" + integer);
                        }
                )
                .log()
                .subscribe();
    }

    /**
     * Flux.create 与 Flux.generate 的主要区别如下：
     * 1、数据生成方式：
     * Flux.generate：同步地逐个生成数据，每次只能发出一个元素
     * Flux.create：异步地生成数据，可以在不同的线程中发出多个元素
     * 2、状态管理：
     * Flux.generate：强制要求有状态管理，必须返回下一个状态
     * Flux.create：不需要状态管理，更加灵活
     * 3、使用场景：
     * Flux.generate：适用于简单的、基于状态的序列生成
     * Flux.create：适用于将现有的异步API（如监听器、回调）桥接到响应式流
     * 4、背压处理：
     * Flux.generate：由Reactor自动处理背压
     * Flux.create：需要开发者考虑背压策略
     * 5、线程模型：
     * Flux.generate：在同一个调用栈中同步执行
     * Flux.create：支持多线程环境，可以跨线程发出数据
     */

    /**
     * 多线程
     */
    public static void create() {
        /**
         * FluxSink 是 Project Reactor 中用于异步数据发射的重要接口，其主要作用如下：
         * 1、异步数据发射：
         * 允许在不同线程中发射数据
         * 适用于将现有异步API桥接到响应式流的场景
         * 可以在回调、监听器等异步环境中使用
         * 2、生命周期管理：
         * 提供 onDispose() 和 onCancel() 方法来注册清理回调
         * 可以检测订阅是否被取消或 disposed
         * 3、背压处理策略：
         * 支持多种背压策略（IGNORE、BUFFER、DROP、LATEST、ERROR）
         * 可以通过 overflow() 方法处理背压情况
         * 4、主要使用场景：
         * 在 Flux.create() 方法中使用
         * 适用于事件驱动的异步数据源
         * 将传统的回调/监听器模式转换为响应式流
         * 5、与 SynchronousSink 的区别：
         * SynchronousSink 只能在同步上下文中使用
         * FluxSink 可以在异步环境中使用
         */
        Flux.create(fluxSink -> {
                    MyListener listener = new MyListener(fluxSink);
                    for (int i = 0; i < 10; i++) {
                        listener.online("用户" + i);
                    }
                })
                .log()
                .subscribe();
    }

    public static void dispose() {
        Flux<String> flux = Flux.range(1, 10000)
                .delayElements(Duration.ofSeconds(1))
                .map(v -> "数据：" + v)
                .log();
        /**
         * 在 Project Reactor 中，Disposable 接口的作用是表示一个可以被取消或 dispose 的操作或资源。
         * 具体作用包括：
         * 1、资源管理：
         *  Disposable 允许你管理订阅资源，可以在不需要时主动取消订阅
         *  防止内存泄漏和不必要的资源消耗
         * 2、取消订阅：
         *  通过调用 dispose() 方法可以主动取消正在进行的订阅
         *  一旦调用 dispose()，数据流会停止，不再接收新的元素
         * 3、生命周期控制：
         *  提供对响应式流生命周期的控制能力
         *  可以在适当的时候清理资源
         */
        Disposable subscribe = flux.subscribe(source -> System.out.println("数据：" + source));
        new Thread(() -> {
            try {
                Thread.sleep(10000);
                subscribe.dispose();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }).start();
    }

    /**
     * 单线程
     */
    public static void generate() {
        /**
         * Flux.generate() 是一个用于编程方式生成数据流的工厂方法，主要作用包括：
         * 1、程序化生成数据：
         *  允许通过编程方式逐个生成数据元素
         *  提供对数据生成过程的精确控制
         * 2、状态管理：
         *  支持维护生成过程中的状态
         *  可以基于前一个状态生成下一个元素
         * 3、终止控制：
         *  可以在适当的时候发出完成信号
         *  支持条件性终止数据流
         */
        Flux.generate(
                        /**
                         * 初始值
                         */
                        () -> 0,
                        /**
                         * 生成数据
                         */
                        (state, synchronousSink) -> {
                            if (state == 10) {
                                /**
                                 * 完成信号
                                 */
                                synchronousSink.complete();
                            }
                            if (state == 7) {
                                /**
                                 * 异常信号
                                 */
                                synchronousSink.error(new RuntimeException("出错了"));
                            }
                            /**
                             * 下一个元素
                             */
                            synchronousSink.next(state);
                            /**
                             * 下一个状态
                             */
                            return state + 1;
                        }
                )
                .doOnError(e -> System.out.println("出错了：" + e))
                .log()
                .subscribe();
    }

    public static void limit() {
        Flux.range(1, 10)
                .log()
                /**
                 * limitRate(3) 的作用是限制上游发布者向下游发送数据的速率，实现背压控制。
                 * 具体来说：
                 * limitRate(3) 设置了一个请求窗口大小为3，这意味着下游每次最多只能从上游请求3个元素
                 * 当下游处理完这3个元素后，会自动请求下一批3个元素
                 * 这种机制防止了上游发送数据过快而导致下游处理不过来的情况
                 */
                .limitRate(3)
//                .log()
                .subscribe(System.out::println);
    }

    public static void buffer() {
        Flux.range(1, 9)
                .log()
                /**
                 * 缓冲区：将上游 Flux 发出的元素每3个分为一组，形成一个 List 集合
                 */
                .buffer(3)
                .log()
                .subscribe(list -> {
                    list.forEach(System.out::println);
                });

        Flux.range(1, 9)
                .buffer(3)
                .subscribe(new BaseSubscriber<List<Integer>>() {
                    @Override
                    protected void hookOnSubscribe(Subscription subscription) {
                        request(1);
                    }

                    @Override
                    protected void hookOnNext(List<Integer> value) {
                        for (Integer integer : value) {
                            System.out.println("数据：" + integer);
                        }
                        request(1);
                    }
                });
    }

}
